package com.kusen.village.raise.config;

import com.kusen.village.raise.service.MqttReceiveService;
import com.kusen.village.raise.service.impl.MqttReceiveServiceImpl;
import com.kusen.village.raise.util.StringUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;

/**
 * MQTT接收消息处理
 * 获取硬件向mqtt发送数据时，同时服务器立马获取到数据。
 * 1.插入数据库
 * 2.发送警告信息
 *
 * @author ghl
 */
@Configuration
@IntegrationComponentScan
public class MqttReceiveConfig {
    public MqttReceiveConfig() {
        setMqttReceiveService(new MqttReceiveServiceImpl());
    }

    @Value("${spring.mqtt.client.id}")
    private String clientId;

    @Value("${spring.mqtt.default.topic}")
    private String defaultTopic;

    @Value("${spring.mqtt.completionTimeout}")
    private int completionTimeout;   //连接超时
    @Autowired
    private MqttPahoClientFactory mqttClientFactory;

    //接收通道
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    //配置client,监听的topic
    @Bean
    public MessageProducer inbound() {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(clientId + StringUtil.serialNumber("SUB"), mqttClientFactory,
                        "hopeful", "oit", "aa", "bb");
        adapter.setCompletionTimeout(completionTimeout);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    /**
     * 接收topic的数据
     */
    private MqttReceiveService mqttReceiveService;

    public void setMqttReceiveService(MqttReceiveService mqttReceiveService) {
        this.mqttReceiveService = mqttReceiveService;
    }

    /**
     * 通过通道获取数据
     *
     * @return
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                String topic = message.getHeaders().get("mqtt_receivedTopic").toString();
                mqttReceiveService.handlerMqttMessage(topic, message.getPayload().toString());
            }
        };
    }

    /**
     * 通道2
     *
     * @return
     */
    @Bean
    public MessageChannel mqttInputChannelTwo() {
        return new DirectChannel();
    }

    /**
     * 配置client2，监听的topic:hell2,hello3
     *
     * @return
     */
    @Bean
    public MessageProducer inbound1() {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(clientId + "_inboundTwo_" + StringUtil.serialNumber("SUB"), mqttClientFactory,
                        "iot", "hello3", "docker");
        adapter.setCompletionTimeout(completionTimeout);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannelTwo());
        return adapter;
    }

    /**
     * 通过通道2获取数据
     *
     * @return
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannelTwo")
    public MessageHandler handlerTwo() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                String topic = message.getHeaders().get("mqtt_receivedTopic").toString();
                mqttReceiveService.handlerMqttMessage(topic, message.getPayload().toString());
            }
        };
    }
}
